package com.danan.realtime.function;

import com.alibaba.fastjson.JSONObject;
import com.danan.realtime.util.JDBCUtil;
import com.danan.realtime.util.RedisUtil;
import com.danan.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;


import java.sql.Connection;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author NanHuang
 * @Date 2023/1/26
 */
public abstract class MyAsyncFunction<T> extends RichAsyncFunction<T,T> {

    public abstract String getTable();
    public abstract String getId(T input);
    protected abstract void extendDimInfo(T data, JSONObject dimInfo);
    private ThreadPoolExecutor threadPool;

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    // 1 获取每个线程的客户端
                    Connection phoenixConn = JDBCUtil.getPhoenixConnection();
                    Jedis jedis = RedisUtil.getJedis();
                    // 2 获取配置信息
                    JSONObject dimInfo = JDBCUtil.getDimInfoFromPhoenix(jedis, phoenixConn, getTable(), getId(input));
                    // 3 完善配置信息
                    extendDimInfo(input,dimInfo);
                    // 4 收集结果
                    resultFuture.complete(Collections.singletonList(input));
                    // 5 关闭连接
                    phoenixConn.close();
                    jedis.close();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }

            }
        });
    }
}
